nanopyx.__agent__

  1import platform
  2import random
  3
  4import numpy as np
  5from sklearn.linear_model import LogisticRegression
  6from scipy.stats import norm
  7
  8from .liquid.__njit__ import njit_works
  9from .liquid.__opencl__ import opencl_works, devices
 10
 11class Agent_:
 12
 13    """
 14    Base class for the Agent of the Nanopyx Liquid Engine 
 15    Pond, James Pond
 16    """
 17
 18    def __init__(self,) -> None:
 19        """
 20        Initialize the Agent
 21        The agent is supposed to work as a singleton object, initialized only once in the __init__.py of nanopyx
 22        PS: (Is this good enough or is it necessary to implement the singleton design pattern?)
 23
 24        Agent responsabilities:
 25            1. Store the current state of the machine (e.g. OS, CPU, RAM, GPU, Python version etc.);
 26            2. Store the current state of ALL initialized LE objects (e.g. anything that is currently running, anything that is scheduled to run,
 27                runs previously executed in the current session etc.);
 28            3. Whenever a LE object wants to run, it must query the Agent on what is the best implementation for it;
 29            4. Tests whether there was an unexpected delay and adjust following paths based on it;
 30        """
 31
 32        ### MACHINE INFO ###
 33        self.os_info = {'OS':platform.platform(),'Architecture':platform.machine()}
 34        self.cpu_info = {'CPU':platform.processor()}
 35        self.ram_info = {'RAM':'TBD'}
 36        self.py_info = {'Version':platform.python_version(),'Implementation':platform.python_implementation(),'Compiler':platform.python_compiler()}
 37
 38        self.numba_info = {'Numba':njit_works()}
 39        self.pyopencl_info = {'PyOpenCL':opencl_works(),'Devices':devices}
 40        self.cuda_info = {'CUDA':'TBD'}
 41        ### MACHINE INFO ###
 42
 43        self._current_runs = []
 44        self._scheduled_runs = []
 45        self._finished_runs = []
 46        
 47        self.delayed_runtypes = {}  # Store runtypes as keys and their values as (delay_factor, delay_prob)
 48
 49    def _get_ordered_run_types(self, fn, args, kwargs):
 50        """
 51        Retrieves an ordered list of run_types for the given args and kwargs
 52        """
 53
 54        # str representation of the arguments and their corresponding 'norm'
 55        repr_args, repr_norm = fn._get_args_repr_score(*args, **kwargs)
 56        # dictionary to hold speeds
 57        fast_avg_speed = {}
 58        fast_std_speed = {}
 59        slow_avg_speed = {}
 60        slow_std_speed = {}
 61        # fn._benchmarks is a dictionary of dictionaries. The first key is the run_type, the second key is the repr_args
 62        # Check every run_type for the most similar args
 63        for run_type in fn._run_types:
 64            if repr_args in fn._benchmarks[run_type]:
 65                run_info = fn._benchmarks[run_type][repr_args][1:]
 66            else:
 67                # if the repr_args are not in the benchmarks, find the most similar repr_args
 68                best_score = np.inf
 69                best_repr_args = None
 70                for repr_args_ in fn._benchmarks[run_type]:
 71                    score = np.abs(fn._benchmarks[run_type][repr_args_][0] - repr_norm)
 72                    if score < best_score:
 73                        best_score = score
 74                        best_repr_args = repr_args_
 75                # What happens if there are no benchmarks for this runtype?
 76                if best_repr_args is None:
 77                    run_info = [None] 
 78                else:
 79                    run_info = fn._benchmarks[run_type][best_repr_args][1:]
 80
 81            if len(run_info)<2:
 82                # Fall back to default values
 83                if 'OpenCL' in run_type:
 84                    rt = 'OpenCL'
 85                else:
 86                    rt = run_type
 87
 88                best_score = np.inf
 89                best_repr_args = None
 90                for repr_args_ in fn._default_benchmarks[rt]:
 91                    score = np.abs(fn._default_benchmarks[rt][repr_args_][0] - repr_norm)
 92                    if score < best_score:
 93                        best_score = score
 94                        best_repr_args = repr_args_
 95                run_info = fn._default_benchmarks[rt][best_repr_args][1:]
 96
 97            run_info = np.array(run_info)
 98            if len(run_info)>50:
 99                run_info = run_info[-50:]
100
101            fast_values = np.partition(run_info,len(run_info)//2)[:len(run_info)//2]
102            slow_values = np.partition(run_info,len(run_info)//2)[len(run_info)//2:]
103            fast_avg_speed[run_type] = np.average(fast_values)
104            fast_std_speed[run_type] = np.std(fast_values)
105            slow_avg_speed[run_type] = np.average(slow_values)
106            slow_std_speed[run_type] = np.std(slow_values)
107
108        return fast_avg_speed, fast_std_speed, slow_avg_speed, slow_std_speed
109    
110    def _calculate_prob_of_delay(self, runtimes_history, avg, std):
111        """
112        Calculates the probability that the given run_type is still delayed using historical data
113        """
114
115        # Boolean array, True if delay, False if not
116        delays = runtimes_history > avg+4*std
117
118        model = LogisticRegression()
119        model.fit([[state] for state in delays[:-1]], delays[1:])
120        
121        return model.predict_proba([[True]])[:,model.classes_.tolist().index(True)][0]
122
123    def _check_delay(self, run_type, runtime, runtimes_history):
124        """
125        Checks if the given run_type ran delayed in the previous run when compared with historical data
126        If delayed:
127            1. Calculates a probability that this delay is maintained
128            2. Stores the delay factor and the probability
129        """
130        
131        threaded_runtypes = ["Threaded", "Threaded_static", "Threaded_dynamic", "Threaded_guided"]
132        
133        runtimes_history = np.array(runtimes_history)
134        if len(runtimes_history)>50:
135            runtimes_history = runtimes_history[-50:]
136        fast_values = np.partition(runtimes_history,len(runtimes_history)//2)[:len(runtimes_history)//2]
137        slow_values = np.partition(runtimes_history,len(runtimes_history)//2)[len(runtimes_history)//2:]
138        
139        fast_avg_speed = np.average(fast_values)
140        fast_std_speed = np.std(fast_values)
141        slow_avg_speed = np.average(slow_values)
142        slow_std_speed = np.std(slow_values)
143
144        if run_type in self.delayed_runtypes:
145            if runtime < (slow_avg_speed - slow_std_speed) or runtime < (fast_avg_speed + fast_std_speed):
146                if "Threaded" in run_type:
147                    for threaded_run_type in threaded_runtypes:
148                        self.delayed_runtypes.pop(threaded_run_type, None)
149                else:
150                    if run_type in self.delayed_runtypes:
151                        self.delayed_runtypes.pop(run_type, None)
152                return 'Delay off'
153    
154        if runtime > fast_avg_speed + 4*fast_std_speed:
155            runtimes_history = np.append(runtimes_history,runtime)
156            delay_factor = runtime / fast_avg_speed
157            try:
158                delay_prob = self._calculate_prob_of_delay(runtimes_history, fast_avg_speed, fast_std_speed)
159            except ValueError:
160                delay_prob = 0.01
161            print(f"Run type {run_type} was delayed in the previous run. Delay factor: {delay_factor}, Delay probability: {delay_prob}")
162
163            if "Threaded" in run_type:
164                for threaded_run_type in threaded_runtypes:
165                    self.delayed_runtypes[threaded_run_type] = (delay_factor, delay_prob)
166            else:
167                self.delayed_runtypes[run_type] = (delay_factor, delay_prob)
168
169    
170    def _adjust_times(self, fast_device_times, slow_device_times):
171        """
172        Adjusts the historic avg time of a run_type if it was delayed in previous runs
173        """
174        adjusted_times = fast_device_times.copy()
175        for runtype in self.delayed_runtypes.keys():
176            if runtype in fast_device_times.keys():
177                delay_factor, delay_prob = self.delayed_runtypes[runtype]
178                # Weighted avg by the probability the run_type is still delayed
179                # expected_time * P(~delay) + delayed_time * P(delay)
180                adjusted_times[runtype] = fast_device_times[runtype] * (1 - delay_prob) + fast_device_times[runtype] * delay_factor * delay_prob
181
182        return adjusted_times
183
184    def get_run_type(self, fn, args, kwargs):
185        """
186        Returns the best run_type for the given args and kwargs
187        """
188        
189        # Get list of run types
190        fast_avg, fast_std, slow_avg, slow_std = self._get_ordered_run_types(fn, args, kwargs)
191        
192        # Penalize the average time a run_type had if that run_type was delayed in previous runs
193        if len(self.delayed_runtypes.keys()) > 0:
194            adjusted_avg = self._adjust_times(fast_avg, slow_avg)
195
196            if sorted(fast_avg, key=fast_avg.get)[0] == sorted(adjusted_avg, key=adjusted_avg.get)[0]:
197                return sorted(fast_avg, key=fast_avg.get)[0]
198
199            weights = [(1/adjusted_avg[k])**2 for k in adjusted_avg]
200            weights = weights / np.sum(weights)
201            
202            # failsafe
203            if sum(weights) == 0:
204                weights = [1 for k in adjusted_avg]
205                
206            return random.choices(list(adjusted_avg.keys()), weights=weights, k=1)[0]
207        else:
208            return sorted(fast_avg, key=fast_avg.get)[0]
209        
210
211    def _inform(self, fn):
212        """
213        Informs the Agent that a LE object finished running
214        """
215
216        repr_args = fn._last_args
217        run_type = fn._last_runtype
218        
219        historical_data = fn._benchmarks[run_type][repr_args][1:]
220        
221        assert historical_data[-1] == fn._last_time, "Historical data is not consistent with the last runtime"
222
223        print(f"Agent: {fn._designation} using {run_type} ran in {fn._last_time} seconds")
224
225        if len(historical_data) > 19:
226            self._check_delay(run_type, historical_data[-1], historical_data[:-1])
227
228
229Agent = Agent_()